package sunyu.kit.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.client.coprocessor.AggregationClient;
import org.apache.hadoop.hbase.client.coprocessor.LongColumnInterpreter;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.mapreduce.Job;
import org.slf4j.Logger;
import sunyu.kit.common.ExceptionKit;
import sunyu.kit.common.LogKit;
import sunyu.kit.common.PageKit;
import sunyu.kit.common.StringKit;
import sunyu.kit.hbase.mapper.pojo.ColumnFamily;
import sunyu.kit.hbase.mapper.pojo.RowData;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * hbase工具类
 *
 * @author 孙宇
 */
public class HbaseKit {

    public static final String aggregateImplementationCoprocessor = "org.apache.hadoop.hbase.coprocessor.AggregateImplementation";//统计用的协处理器类名称
    private static final Logger logger = LogKit.getLogger();
    private static final LongColumnInterpreter longColumnInterpreter = new LongColumnInterpreter();
    private static Configuration config;//hbase配置
    private static Connection connection;//hbase链接
    private static AggregationClient aggregationClient;//协处理器客户端

    /**
     * 初始化Hbase
     *
     * @param quorum zookeeper地址
     * @param znode  节点名称，默认是/hbase
     */
    public static void init(String quorum, String znode) {
        if (connection == null) {
            logger.info("初始化Hbase开始");
            logger.info("hbase.zookeeper.quorum：{}", quorum);
            logger.info("zookeeper.znode.parent：{}", znode);

            config = HBaseConfiguration.create();
            config.set("hbase.zookeeper.quorum", quorum);
            config.set("zookeeper.znode.parent", znode);

            try {
                connection = ConnectionFactory.createConnection(config);
            } catch (IOException e) {
                logger.error(ExceptionKit.getStackTrace(e));
            }

            aggregationClient = new AggregationClient(config);
            logger.info("初始化Hbase结束");
        } else {
            logger.error("hbase已经初始化");
        }
    }

    /**
     * 初始化Hbase
     *
     * @param quorum           zookeeper地址
     * @param znode            节点名称，默认是/hbase
     * @param rpcTimeout
     * @param operationTimeout
     * @param scannerTimeout
     */
    public static void init(String quorum, String znode, String rpcTimeout, String operationTimeout, String scannerTimeout, String ipcTimeout) {
        if (StringKit.isBlank(rpcTimeout)) {
            rpcTimeout = "60000";
        }
        if (StringKit.isBlank(operationTimeout)) {
            operationTimeout = "120000";
        }
        if (StringKit.isBlank(scannerTimeout)) {
            scannerTimeout = "60000";
        }
        if (StringKit.isBlank(ipcTimeout)) {
            ipcTimeout = "60000";
        }
        if (connection == null) {
            logger.info("初始化Hbase开始");
            logger.info("hbase.zookeeper.quorum：{}", quorum);
            logger.info("zookeeper.znode.parent：{}", znode);
            logger.info("hbase.rpc.timeout：{}", rpcTimeout);
            logger.info("hbase.client.operation.timeout：{}", operationTimeout);
            logger.info("hbase.client.scanner.timeout.period：{}", scannerTimeout);
            logger.info("ipc.client.connect.timeout：{}", ipcTimeout);

            config = HBaseConfiguration.create();
            config.set("hbase.zookeeper.quorum", quorum);
            config.set("zookeeper.znode.parent", znode);
            config.set("hbase.rpc.timeout", rpcTimeout);
            config.set("hbase.client.operation.timeout", operationTimeout);
            config.set("hbase.client.scanner.timeout.period", scannerTimeout);
            config.set("ipc.client.connect.timeout", ipcTimeout);

            try {
                connection = ConnectionFactory.createConnection(config);
            } catch (IOException e) {
                logger.error(ExceptionKit.getStackTrace(e));
            }

            aggregationClient = new AggregationClient(config);
            logger.info("初始化Hbase结束");
        } else {
            logger.error("hbase已经初始化");
        }
    }

    /**
     * 系统关闭前，调用此方法，销毁链接
     */
    public static void destroy() {
        logger.info("销毁Hbase连接池开始");
        if (!connection.isClosed()) {
            try {
                connection.close();
            } catch (IOException e) {
                logger.error(ExceptionKit.getStackTrace(e));
            }
        }
        try {
            aggregationClient.close();
        } catch (IOException e) {
            logger.error(ExceptionKit.getStackTrace(e));
        }
        logger.info("销毁Hbase连接池结束");
    }

    /**
     * 获得Admin
     *
     * @param connection
     * @return
     */
    private static Admin getAdmin(Connection connection) {
        try {
            return connection.getAdmin();
        } catch (IOException e) {
            logger.error(ExceptionKit.getStackTrace(e));
        }
        return null;
    }

    /**
     * 创建表
     *
     * @param tableName   表名
     * @param familyNames 列族名称集合
     */
    public static boolean createTable(String tableName,
                                      String[] familyNames) {
        try (Admin admin = getAdmin(connection)) {
            if (admin.tableExists(TableName.valueOf(tableName))) {
                logger.error("{} 表已存在!", tableName);
            } else {
                HTableDescriptor hTableDescriptor = new HTableDescriptor(TableName.valueOf(tableName));
                hTableDescriptor.addCoprocessor(HbaseKit.aggregateImplementationCoprocessor);
                for (String familyName : familyNames) {
                    HColumnDescriptor family = new HColumnDescriptor(familyName);
                    hTableDescriptor.addFamily(family);
                }
                admin.createTable(hTableDescriptor);
                return true;
            }
        } catch (IOException e) {
            logger.error(ExceptionKit.getStackTrace(e));
        }
        return false;
    }

    /**
     * 添加协处理器
     *
     * @param tableName 表名
     * @param className 协处理器类名
     */
    public static boolean addCoprocessor(String tableName,
                                         String className) {
        try (Admin admin = getAdmin(connection); Table table = connection.getTable(TableName.valueOf(tableName))) {
            if (admin.tableExists(TableName.valueOf(tableName))) {
                HTableDescriptor desc = table.getTableDescriptor();
                if (!desc.getCoprocessors().contains(className)) {
                    try {
                        admin.disableTable(TableName.valueOf(tableName));//先禁用表
                        HTableDescriptor hTableDescriptor = admin.getTableDescriptor(TableName.valueOf(tableName));//获得表描述
                        hTableDescriptor.addCoprocessor(className);//添加协处理器
                        admin.modifyTable(TableName.valueOf(tableName), hTableDescriptor);//修改表
                    } catch (IOException e) {
                        logger.error(ExceptionKit.getStackTrace(e));
                    } finally {
                        try {
                            admin.enableTable(TableName.valueOf(tableName));//启用表
                        } catch (IOException e) {
                            logger.error(ExceptionKit.getStackTrace(e));
                        }
                    }
                } else {
                    logger.error("{} 协处理器已存在!", className);
                }
            } else {
                logger.error("{} 表不存在!", tableName);
            }
        } catch (IOException e) {
            logger.error(ExceptionKit.getStackTrace(e));
        }
        return false;
    }

    /**
     * 删除表
     *
     * @param tableName 表名
     */
    public static boolean deleteTable(String tableName) {
        try (Admin admin = getAdmin(connection)) {
            if (admin.tableExists(TableName.valueOf(tableName))) {
                admin.disableTable(TableName.valueOf(tableName));
                admin.deleteTable(TableName.valueOf(tableName));
                return true;
            } else {
                logger.error("{} 表不存在!", tableName);
            }
        } catch (IOException e) {
            logger.error(ExceptionKit.getStackTrace(e));
        }
        return false;
    }

    /**
     * 获得table描述
     *
     * @param tableName 表名
     * @return
     */
    public static HTableDescriptor tableDescriptor(String tableName) {
        try (Table table = connection.getTable(TableName.valueOf(tableName))) {
            return table.getTableDescriptor();
        } catch (IOException e) {
            logger.error(ExceptionKit.getStackTrace(e));
        }
        return null;
    }

    /**
     * 插入一行
     *
     * @param tableName 表名
     * @param put
     */
    public static void put(String tableName,
                           Put put) {
        put(tableName, Arrays.asList(put));
    }

    /**
     * 插入多行
     *
     * @param tableName 表名
     * @param puts
     */
    public static void put(String tableName,
                           List<Put> puts) {
        try (Table table = connection.getTable(TableName.valueOf(tableName))) {
            table.put(puts);
        } catch (IOException e) {
            logger.error(ExceptionKit.getStackTrace(e));
        }
    }

    /**
     * 扫描
     *
     * @param tableName 表名
     * @return
     */
    public static List<Result> scan(String tableName) {
        return scan(tableName, new Scan());
    }

    /**
     * 扫描
     *
     * @param tableName 表名
     * @param pageSize  查询多少条记录
     * @return
     */
    public static List<Result> scan(String tableName,
                                    Integer pageSize) {
        return scan(tableName, new Scan(), pageSize);
    }

    /**
     * 扫描
     *
     * @param tableName 表名
     * @param scan
     * @return
     */
    public static List<Result> scan(String tableName,
                                    Scan scan) {
        return scan(tableName, scan, null);
    }

    /**
     * 扫描
     *
     * @param tableName 表名
     * @param scan
     * @param pageSize  查询多少条记录
     * @return
     */
    public static List<Result> scan(String tableName,
                                    Scan scan,
                                    Integer pageSize) {
        return scan(tableName, scan, pageSize, null);
    }

    /**
     * 扫描
     *
     * @param tableName  表名
     * @param scan
     * @param pageSize   查询多少条记录
     * @param pageNumber 扫描第几页，页数越大，查询越慢，因为要再内存中翻到这一页
     * @return
     */
    public static List<Result> scan(String tableName,
                                    Scan scan,
                                    Integer pageSize,
                                    Integer pageNumber) {
        List<Result> results = new ArrayList<>();
        try (Table table = connection.getTable(TableName.valueOf(tableName)); ResultScanner resultScanner = table.getScanner(scan)) {
            if (pageSize != null) {
                if (pageNumber != null && pageNumber > 1) {//说明要返回第pageNumber页的内容
                    AtomicLong i = new AtomicLong(0);
                    for (Result result : resultScanner) {
                        if (i.incrementAndGet() > PageKit.getOffset(pageSize, pageNumber)) {
                            results.add(result);
                        }
                        if (i.get() == pageSize * pageNumber) {
                            break;
                        }
                    }
                } else {
                    Result[] rs = resultScanner.next(pageSize);
                    for (Result result : rs) {
                        results.add(result);
                    }
                }
            } else {
                for (Result result : resultScanner) {
                    results.add(result);
                }
            }
        } catch (IOException e) {
            logger.error(ExceptionKit.getStackTrace(e));
        }
        return results;
    }

    /**
     * 删除一行
     *
     * @param tableName 表名
     * @param del
     */
    public static boolean delete(String tableName,
                                 Delete del) {
        try (Table table = connection.getTable(TableName.valueOf(tableName))) {
            table.delete(del);
            return true;
        } catch (IOException e) {
            logger.error(ExceptionKit.getStackTrace(e));
        }
        return false;
    }

    /**
     * 获得一行
     *
     * @param tableName 表名
     * @param g
     * @return
     */
    public static Result get(String tableName,
                             Get g) {
        Result[] result = get(tableName, Arrays.asList(g));
        if (result != null && result.length > 0) {
            return result[0];
        }
        return null;
    }

    /**
     * 获得多行
     *
     * @param tableName 表名
     * @param gets
     * @return
     */
    public static Result[] get(String tableName,
                               List<Get> gets) {
        try (Table table = connection.getTable(TableName.valueOf(tableName))) {
            return table.get(gets);
        } catch (IOException e) {
            logger.error(ExceptionKit.getStackTrace(e));
        }
        return null;
    }

    /**
     * 统计行数，依赖协处理器，如果统计出错，返回null
     * <p>
     * 请注意，scan如果有startRow和stopRow，那么startRow一定要小于stopRow，否则会抛异常
     *
     * @param tableName 表名
     * @param scan
     * @return
     */
    public static Long rowCount(String tableName,
                                Scan scan) throws Throwable {
        return aggregationClient.rowCount(TableName.valueOf(tableName), HbaseKit.longColumnInterpreter, scan);
    }

    /**
     * result转换row对象
     *
     * @param result             hbase查询出的结果
     * @param returnRowKeyOnly   是否只看rowkey
     * @param showHistoryVersion 是否转换历史版本数据(需要scan中scan.setMaxVersions(Integer.MAX_VALUE);或者get.setMaxVersions(Integer.MAX_VALUE);)
     * @return
     */
    public static RowData mapRow(Result result,
                                 Boolean returnRowKeyOnly,
                                 Boolean showHistoryVersion) {
        List<RowData> rowDatas = mapRow(Arrays.asList(result), returnRowKeyOnly, showHistoryVersion);
        if (rowDatas != null && rowDatas.size() > 0) {
            return rowDatas.get(0);
        }
        return null;
    }

    /**
     * result转换row对象
     *
     * @param results            hbase查询出的结果
     * @param returnRowKeyOnly   是否只看rowkey
     * @param showHistoryVersion 是否转换历史版本数据(需要scan中scan.setMaxVersions(Integer.MAX_VALUE);或者get.setMaxVersions(Integer.MAX_VALUE);)
     * @return
     */
    public static List<RowData> mapRow(List<Result> results,
                                       Boolean returnRowKeyOnly,
                                       Boolean showHistoryVersion) {
        if (results != null && results.size() > 0) {
            List<RowData> rowDatas = new ArrayList<>();
            for (Result result : results) {
                if (result == null || result.getRow() == null) {
                    return null;
                }
                RowData rowData = new RowData();// 一行
                rowData.setRowName(Bytes.toString(result.getRow()));
                if (returnRowKeyOnly == null || returnRowKeyOnly == false) {//如果没传只返回rowkey的参数，或者这个参数是false
                    Map<String, ColumnFamily> familys = rowData.getFamilys();// 一行里的family集合
                    for (Cell cell : result.rawCells()) {
                        String familyName = Bytes.toString(CellUtil.cloneFamily(cell));// 获取当前列的familyName
                        String columnName = Bytes.toString(CellUtil.cloneQualifier(cell));// 列名
                        String value = Bytes.toString(CellUtil.cloneValue(cell));// 值

                        ColumnFamily columnFamily = familys.get(familyName);
                        if (columnFamily == null) {
                            columnFamily = new ColumnFamily();
                            familys.put(familyName, columnFamily);
                        }

                        Map<String, Object> columns = columnFamily.getColumns();// 当前family的所有列
                        columns.put(columnName, value);
                    }

                    //多版本的时候
                    if (showHistoryVersion != null && showHistoryVersion == true) {
                        for (Cell c : result.listCells()) {
                            String familyName = Bytes.toString(CellUtil.cloneFamily(c));// 获取当前列的familyName
                            String columnName = Bytes.toString(CellUtil.cloneQualifier(c));// 列名
                            String value = Bytes.toString(CellUtil.cloneValue(c));// 值
                            long timestamp = c.getTimestamp();//插入时间

                            ColumnFamily columnFamily = familys.get(familyName);
                            Map<String, Object> historyColumns = columnFamily.getHistoryColumns();

                            List columnVersionList = (List) historyColumns.get(columnName);//当前列的多版本map
                            if (columnVersionList == null) {
                                columnVersionList = new ArrayList();
                                historyColumns.put(columnName, columnVersionList);
                            }

                            Map<String, Object> currentColumn = new HashMap<String, Object>();
                            currentColumn.put("value", value);
                            currentColumn.put("timestamp", timestamp);
                            columnVersionList.add(currentColumn);
                        }
                    }
                }
                rowDatas.add(rowData);
            }
            return rowDatas;
        }
        return new ArrayList();
    }

    public static Job getJob(String jobName) {
        Job job = null;
        try {
            job = Job.getInstance(config, jobName);
        } catch (IOException e) {
            logger.error(ExceptionKit.getStackTrace(e));
        }
        return job;
    }

}